package rpc.remote.transport.netty.client;


import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;

import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import rpc.remote.dto.RpcRequest;
import rpc.remote.dto.RpcResponse;
import rpc.serialize.kyro.KryoSerializer;
import rpc.remote.transport.netty.code.NettyKryoDecoder;
import rpc.remote.transport.netty.code.NettyKryoEncoder;


/**
 * @ClassName: NettyClient
 * @Description:
 * @author: Kowk
 * @since:1.0.0
 */
@Slf4j
public class NettyClient {
    private final String host;
    private final int port;
    private static final Bootstrap b;

    public NettyClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    //静态方法初始化相关资源
    static {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        b = new Bootstrap();
        KryoSerializer kryoSerializer = new KryoSerializer();
        b.group(eventLoopGroup)
                .channel(NioSocketChannel.class)
                .handler(new LoggingHandler(LogLevel.INFO))
                //如果连接超过设定的时间则直接连接失败，如果15秒没有发送数据给服务端则发送一次心跳请求，保证连接正常
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,5000)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //自定义序列化编码解码器
                        //RpcResponse -> Byte 响应信息序列化
                        socketChannel.pipeline().addLast(new NettyKryoDecoder(kryoSerializer, RpcResponse.class));
                        //ByteBuf -> RpcRequest
                        socketChannel.pipeline().addLast(new NettyKryoEncoder(kryoSerializer, RpcRequest.class));
                        socketChannel.pipeline().addLast(new NettyClientHandler());
                    }
                });
    }

    /**
     * @Description: 发送消息到服务端
     * @Param: [request 所发送的消息]
     * @return: rpc.remote.dto.RpcResponse 服务端的响应消息
     */
    public RpcResponse sendMessage(RpcRequest request){
        try {
            ChannelFuture future = b.connect(host,port).sync();
            log.info("客户端连接 {}",host + ":" + port);
            Channel futureChannel = future.channel();
            log.info("发送消息");
            if (futureChannel != null){
                futureChannel.writeAndFlush(request).addListener(future1 -> {
                    if(future1.isSuccess()){
                        log.info("客户端发送消息：[{}]",request.toString());
                    } else {
                        log.error("发送失败:",future1.cause());
                    }
                });
                //阻塞等待，直到Channel关闭
                futureChannel.closeFuture().sync();
                //将服务端返回的数据取出
                AttributeKey<RpcResponse> key = AttributeKey.valueOf("rpcResponse");
                return futureChannel.attr(key).get();
            }
        } catch (InterruptedException e) {
            log.error("连接客户端的时候发生错误：",e);
        }
        return null;
    }
}
